package com.wz.jiangsu.controller;

import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.wz.jiangsu.bean.entity.wazuh.WazuhSecurityEventSafetyDto;
import com.wz.jiangsu.bean.vo.SecurityEventDataCommonGoWazuhDto;
import com.wz.jiangsu.bean.vo.SecurityEventDataCommonWazuhDto;
import com.wz.jiangsu.bean.vo.resp.R;
import com.wz.jiangsu.bean.vo.show.DealWithEvent;
import com.wz.jiangsu.bean.vo.show.ShowSecurityEventDataCommonWazuhDto;
import com.wz.jiangsu.service.WazuhService;
import com.wz.jiangsu.util.JsonUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.SuccessCallback;
import org.springframework.web.bind.annotation.*;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

/**
 * @program: wz
 * @author: wangzhou
 * @create: 2023-12-13 16:21
 * @description:
 **/
@Slf4j
@RestController
public class WazuhController {


    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Value("${wazuh.topic}")
    private String wazuhTopic;


    @Value("${netvine.security.topic}")
    private String syslog;

    @Autowired
    private WazuhService wazuhService;

    @Autowired
    @Qualifier("taskExecutor")
    private Executor taskExecutor;


    /**
     * 向 wazuh analysisd_queue 推送消息
     *
     * @param dto 模仿 Kafaka 消息体的实体类
     */
    @PostMapping("/kafka/post/callbackOne/wazuh/receive/analysisd_queue")
    public String sendCallbackOneMessageMoney(@RequestBody SecurityEventDataCommonWazuhDto dto, @RequestParam("topic") String topic, Integer size) {

        for (int i = 0; i < size; i++) {
            dto.setUuid(IdUtil.randomUUID());
            String jsonStr = JSONUtil.toJsonStr(dto);
            // 格式化成为一行
            log.info("-------replaceAllBlank---------");
            String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);

            kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();

                    System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });
        }
        return dto.toString() + "发送的topic:" + topic + ",发送的数据大小:" + size;
    }

    /**
     * 向 wauzh sylog系列的主题推送消息
     *
     * @param dto
     * @return
     */
    @PostMapping("/kafka/post/callbackOne/wazuh/receive/syslog_test")
    public String sendCallbackOneMessageMoneyWazuhEngine(@RequestBody ShowSecurityEventDataCommonWazuhDto dto, @RequestParam("topic") String topic, Integer size) {
        for (int j = 0; j < size; j++) {
            long currentTimeMillis = System.currentTimeMillis();
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String timeStr = cn.hutool.core.date.DateUtil.format(new Date(currentTimeMillis), format);
            log.info("时间戳:{},年月日：{}", currentTimeMillis, timeStr);
            dto.setUuid(UUID.randomUUID().toString());
            String jsonStr = JSONUtil.toJsonStr(dto);
            String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);

            kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();

                    System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });

        }
        return dto.toString() + "\"发送的topic:\" + topic + \",发送的数据大小:\" + size";
    }

    /**
     * 向 topic 推送 Kafka 消息DealEvent,指定大小
     *
     * @param dto
     * @return
     */
    @PostMapping("/kafka/post/callbackOne/event/receive/")
    public String sendCallbackOneMessageAllEvent(@RequestBody DealWithEvent dto, @RequestParam("topic") String
            topic, Integer size) {
        for (int i = 0; i < size; i++) {
            String jsonStr = JSONUtil.toJsonStr(dto);
            log.info("发送过来的原始数据是:{}", jsonStr);
//            dto.setUuid(UUID.randomUUID().toString());
            String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);

            kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();
                    log.info("发送消息成功-安全日志，发送的topic:{},partition:{},offset:{}", topic, partition, offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });
        }
        return dto.toString() + "发送的topic:" + topic + ",发送的数据大小:" + size;
    }

    /**
     * 测试 wazuh 使用 系统的时间戳
     *
     * @param dto
     * @param topic
     * @return
     */
    @PostMapping("/kafka/post/callbackOne/wazuh/receive/syslog_test/gets")
    public String sendCallbackOneMessageMoneyWazuhEnginegetTs(@RequestBody ShowSecurityEventDataCommonWazuhDto
                                                                      dto, @RequestParam("topic") String topic) {
        for (int i = 0; i < 10; i++) {
            long currentTimeMillis = System.currentTimeMillis();
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String timeStr = cn.hutool.core.date.DateUtil.format(new Date(currentTimeMillis), format);
            System.out.println("时间戳:" + currentTimeMillis + "," + "年月日:" + timeStr);
            dto.setTs(currentTimeMillis);
            Object payload = dto.getPayload();
            String jsonStr1 = JSONUtil.toJsonStr(payload);
            JSONObject parseObj = JSONUtil.parseObj(jsonStr1);
            String formatted = DateUtil.format(new Date(currentTimeMillis), "yyyy/M/d HH:mm");
            parseObj.set("report_time", formatted);
            Object obj = (Object) parseObj;
            dto.setPayload(obj);
            String jsonStr = JSONUtil.toJsonStr(dto);
//            System.out.println("jsonStr = " + jsonStr);
//            System.out.println("-------replaceAllBlank---------");
            String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);
//            System.out.println(oneRowJson);

            kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
                //成功的回调
                @Override
                public void onSuccess(SendResult<String, Object> success) {
                    // 消息发送到的topic
                    String topic = success.getRecordMetadata().topic();
                    // 消息发送到的分区
                    int partition = success.getRecordMetadata().partition();
                    // 消息在分区内的offset
                    long offset = success.getRecordMetadata().offset();

//                    System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
                }
            }, new FailureCallback() {
                //失败的回调
                @Override
                public void onFailure(Throwable throwable) {
                    System.out.println("发送消息失败1:" + throwable.getMessage());
                }
            });
            try {
                TimeUnit.SECONDS.sleep(1);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
        return "wazuh_test";
    }

    /**
     * 测试 传向 wauzh 的数据 ---- single 使用时间是自带的时间
     *
     * @param dto
     * @param topic
     * @return
     */
    @PostMapping("/kafka/post/callbackOne/wazuh/receive/syslog_test/single")
    public String sendCallbackOneMessageMoneyWazuhEngineSingle(@RequestBody ShowSecurityEventDataCommonWazuhDto
                                                                       dto, @RequestParam("topic") String topic) {

        String jsonStr = JSONUtil.toJsonStr(dto);
//            System.out.println("jsonStr = " + jsonStr);
//            System.out.println("-------replaceAllBlank---------");
        String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);
//            System.out.println(oneRowJson);

        kafkaTemplate.send(topic, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
            //成功的回调
            @Override
            public void onSuccess(SendResult<String, Object> success) {
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();

//                    System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
            }
        }, new FailureCallback() {
            //失败的回调
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("发送消息失败1:" + throwable.getMessage());
            }
        });
        try {
            TimeUnit.SECONDS.sleep(1);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
        return "wazuh_test";
    }

    @GetMapping("/kafka/post/callbackOne/wazuh/receive/timestamp/{timestamp}")
    public String getDateStrByTimestamp(@PathVariable("timestamp") Long timestamp) {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        String timeStr = cn.hutool.core.date.DateUtil.format(new Date(timestamp), format);
//        System.out.println("数据=" + timestamp);
//        System.out.println("传过来的年月日时间:" + timeStr);
        return timeStr;
    }

    /**
     * wauzh吐出来的消息需要给态势
     *
     * @param dto
     * @return
     */
    @PostMapping("/kafka/post/callbackOne/wazuh/send/syslog")
    public String sendCallbackOneMessageMoneyAnalysisdQueue(@RequestBody SecurityEventDataCommonGoWazuhDto dto) {
        String message = StrUtil.toString(dto);
        String jsonStr = JSONUtil.toJsonStr(dto);
        System.out.println("jsonStr = " + jsonStr);

        // 格式化成为一行
        System.out.println("-------replaceAllBlank---------");
        String oneRowJson = JsonUtil.replaceAllBlank(jsonStr);
        System.out.println(oneRowJson);

        kafkaTemplate.send(syslog, oneRowJson).addCallback(new SuccessCallback<SendResult<String, Object>>() {
            //成功的回调
            @Override
            public void onSuccess(SendResult<String, Object> success) {
                // 消息发送到的topic
                String topic = success.getRecordMetadata().topic();
                // 消息发送到的分区
                int partition = success.getRecordMetadata().partition();
                // 消息在分区内的offset
                long offset = success.getRecordMetadata().offset();

                System.out.println("发送消息成功-安全日志:" + topic + "-" + partition + "-" + offset);
            }
        }, new FailureCallback() {
            //失败的回调
            @Override
            public void onFailure(Throwable throwable) {
                System.out.println("发送消息失败1:" + throwable.getMessage());
            }
        });
        return jsonStr;
    }

    /**
     * 查询所有有关 wazuh 数据
     *
     * @return
     */
    @GetMapping("/kafka/post/callbackOne/wazuh/getAllData")
    public R<Map<Integer, WazuhSecurityEventSafetyDto>> getAllMapData() {
        return wazuhService.getAllMapData();
    }

}
